//
// Created by Suzhen on 2017/5/7.
//

#include "BasicEventGenerator.h"

USING_ORION_V2

BasicEventGenerator *BasicEventGenerator::CreateGenerator() {
    return new BasicEventGenerator();
}

void BasicEventGenerator::CloseGenerator(BasicEventGenerator *g) {
    if (!g) {
        return;
    }
    g->stopAsync();
}

void BasicEventGenerator::GeneratorAsyncCb(AsyncContext *async) {
    BasicEventGenerator *gen = static_cast<BasicEventGenerator * >( async->loop->data);
    gen->retain();
    GeneratorLockGuardType guard(gen->internalLock_);
    auto &taskQ = gen->taskQueue_;
    while (!taskQ.empty()) {
        auto &tsk = taskQ.front();
        tsk(*gen);
        taskQ.pop_front();
    }
    gen->release();
}

void BasicEventGenerator::GeneratorAsyncClosedCb(uv_handle_t *h) {
    BasicEventGenerator *gen = static_cast<BasicEventGenerator *>( h->data);
    uv_stop(gen->loopCtx_);
    gen->closePromise_.setValue(YES);
    gen->release();
}

void BasicEventGenerator::submitTask(BasicEventGenerator::TaskType &&task) {
    SYS_FAULT(state_ != CLOSED) << "event generator state error ,expect not CLOSED , current is " << state_;
    GeneratorLockGuardType guard(internalLock_);
    taskQueue_.emplace_back(std::forward<TaskType>(task));
    uv_async_send(asyncCtx_);
}

void BasicEventGenerator::start(std::function<void()> &&initFn) {
    SYS_FAULT(state_ == INIT) << "start invoked, multiple times";
    {
        GeneratorLockGuardType guard(internalLock_);
        state_ = RUNNING;
        threadId_ = getCurrentThreadId();
        //add task
        submitTask([](BasicEventGenerator &o) {
            o.startPromise_.setValue(YES);
        });
    }
    if (initFn) {
        initFn();
    }
    uv_run(loopCtx_, UV_RUN_DEFAULT);
    Utils::CloseUvLoop(loopCtx_);
    state_ = CLOSED;
}

Future<bool> BasicEventGenerator::startAsync(std::function<void()> &&initFn) {
    std::thread th([&, initFn = std::move(initFn)]() mutable {
        start(std::move(initFn));
    });
    th.detach();
    return startFuture();
}


BasicEventGenerator::BasicEventGeneratorState
BasicEventGenerator::getState() {
    return state_;
}

Future<bool> BasicEventGenerator::startFuture() {
    return startPromise_.getFuture();
}

Future<bool> BasicEventGenerator::stopFuture() {
    return closePromise_.getFuture();
}

Future<bool> BasicEventGenerator::stopAsync() {
    if (state_ == CLOSED ||
        state_ == CLOSING ||
        uv_is_closing((uv_handle_t *) asyncCtx_)) {
        return stopFuture();
    }
    {
        GeneratorLockGuardType guard(internalLock_);
        state_ = CLOSING;
    }
    if (getCurrentThreadId() == threadId_) {
        uv_close((uv_handle_t *) asyncCtx_, &GeneratorAsyncClosedCb);
    } else {
        submitTask([](BasicEventGenerator &o) {
            o.stopAsync();
        });
    }
    return stopFuture();
}

CoreLoopContext *BasicEventGenerator::getGeneratorLoopCtx() {
    return loopCtx_;
}

BasicEventGenerator::~BasicEventGenerator() {
    free(loopCtx_);
    free(asyncCtx_);
}

BasicEventGenerator::BasicEventGenerator()
        : MRCSupport(),
          threadId_(),
          loopCtx_(nullptr),
          asyncCtx_(nullptr),
          internalLock_(),
          taskQueue_(),
          closePromise_(),
          startPromise_(),
          state_(INIT) {
    loopCtx_ = (uv_loop_t *) malloc(sizeof(uv_loop_t));
    asyncCtx_ = (uv_async_t *) malloc(sizeof(uv_async_t));
    uv_loop_init(loopCtx_);
    uv_async_init(loopCtx_, asyncCtx_, &GeneratorAsyncCb);
    loopCtx_->data = this;
    asyncCtx_->data = this;
    retain();
}
